package com.atguigu.gmall.realtime.common.util;

/*import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

*//**
 * Connection: Table, Admin
 * Table接口：
 * <p>
 * Table接口用于操作HBase表中的数据。
 * 它提供了用于读写数据的方法，例如get、put、scan、delete等。
 * Table接口允许用户对表进行行级别的数据操作，例如检索、插入或删除特定的行或列。
 * 它是用户与HBase表交互的主要方式，用于执行具体的数据存取操作。
 * Admin接口：
 * <p>
 * Admin接口用于管理HBase的表结构和集群的元数据。
 * 它提供了创建、删除表的方法，以及修改表的属性和结构的方法。
 * Admin接口允许用户查看集群状态、启用或禁用表、修改表的分片（region）等。
 * 它还可以用来获取集群的统计信息和表的元数据信息，例如表的列族信息、表的状态等。
 * Admin接口通常用于表的生命周期管理，而不是用于日常的数据操作。
 *//*
public class HBaseUtil {
    *//**
     * 获取Connection的方法
     *//*
    public static Connection getConnection() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("获取Hbase Connection失败");
        }
    }

    *//**
     * 关闭Connection的方法
     *//*
    public static void closeConnection(Connection connection) {
        if (connection != null && !connection.isClosed()) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException("关闭HBase Connection失败");
            }
        }
    }

    *//**
     * delete
     * @param connection
     * @param namespaceName
     * @param tableName
     * @param rk
     *//*
    *//*public static void deleteRow(Connection connection, String namespaceName, String tableName, String rk){
        Table table = null;
        try {
            TableName tn = TableName.valueOf(namespaceName, tableName);
            table = connection.getTable(tn);

            Delete delete = new Delete(Bytes.toBytes(rk));
            table.delete(delete);
            System.err.println("删除" + namespaceName + ":" + tableName + " 表, rowkey 为 " + rk + " 的数据成功!");

        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException("删除HBase失败");
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }*//*
    public static void deleteRow(Connection connection , String namespaceName, String tableName , String rk ){
        Table table = null ;
        try {
            TableName tn = TableName.valueOf(namespaceName, tableName);
            table = connection.getTable(tn) ;
            Delete delete = new Delete(Bytes.toBytes(rk));
            table.delete(delete);
            System.err.println(" 删除 " + namespaceName+ ":" + tableName + " 表, rowkey为 " + rk+ " 的数据成功");
        }catch ( Exception e ){
            e.printStackTrace();
            throw new RuntimeException(" 删除 Hbase 失败 ");
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    *//**
     * put
     * shell: put 'namespace:table', 'rk', 'cf:cl', 'value'
     * @param connection
     * @param namespaceName
     * @param tableName
     * @param rk
     * @param cf
     * @param data 一条维度数据
     *//*
    *//*public static void putRow(Connection connection, String namespaceName, String tableName, String rk, String cf, JSONObject data){
        // 判空处理
        if (connection == null || namespaceName == null || tableName == null) {
            throw new RuntimeException("参数不能为空");
        }
        Table table = null;
        try {
            // 获取Table对象
            TableName tn = TableName.valueOf(namespaceName, tableName);
            table = connection.getTable(tn);

            Put put = new Put(Bytes.toBytes(rk));
            // 指定列
            data.entrySet()
                    .stream()
                    .filter(entry -> entry.getValue() != null) // 将列值为空的列过滤掉
                    .forEach(entry -> put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(entry.getKey()),  Bytes.toBytes(entry.getValue().toString())));
            table.put(put);
            System.err.println("写入" + namespaceName + ":" + tableName + " 表, rowkey为 " + rk + " 的数据成功!");

        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("写入HBase失败");
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }*//*
    public static void putRow(Connection connection , String namespaceName, String tableName , String rk , String cf , JSONObject data){
        //判空处理
        *//*if (rk==null){
            return;
        }*//*
        Table table = null ;
        try {
            //获取Table对象
            TableName tn = TableName.valueOf(namespaceName, tableName);
            table = connection.getTable(tn);


            Put put = new Put(Bytes.toBytes(rk));
            //指定列
            data.entrySet()
                    .stream()
                    .filter( entry -> entry.getValue() != null )  //将列值为空的列过滤掉
                    .forEach( entry -> put.addColumn(Bytes.toBytes(cf) , Bytes.toBytes(entry.getKey()) , Bytes.toBytes(entry.getValue() != null ? entry.getValue().toString() : " ")));
            table.put( put );
            System.err.println(" 写入 " + namespaceName+ ":" + tableName + " 表, rowkey为 " + rk+ " 的数据成功");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(" 写入 Hbase 失败 ");
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }


    *//**
     * 建表
     *//*
    public static void createTable(Connection connection, String namespaceName, String tableName, String cf) {

        // 非空判断
        if (connection == null || namespaceName == null || tableName == null || cf == null) {
            throw new RuntimeException("参数不能为空");
        }
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            // 1. 判断表是否已经存在
            TableName tn = TableName.valueOf(namespaceName, tableName);
            if (admin.tableExists(tn)){
                System.err.println("建表" + namespaceName + ":" + tableName + ",表已经存在!");
                return;
            }
            // 2. 表不存在,建表
            // 处理列族
            *//*ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf))
                    .build();*//*
            ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes(cf));
            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tn)
                    .setColumnFamily(columnFamilyDescriptor)
                    .build();
            admin.createTable(tableDescriptor);

            System.err.println("建表" + namespaceName + ":" + tableName + ",表创建成功!");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }finally {
            if (admin != null){
                try {
                    admin.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
    *//**
     * 删表
     *//*
    public static void dropTable(Connection connection, String namespaceName, String tableName) {
        // 非空判断
        if (connection == null || namespaceName == null || tableName == null) {
            throw new RuntimeException("参数不能为空");
        }
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            // 1. 判断表是否已经存在
            TableName tn = TableName.valueOf(namespaceName, tableName);
            if (! admin.tableExists(tn)){
                System.err.println("删表" + namespaceName + ":" + tableName + ",表不存在!");
                return;
            }
            // 2. 删表
            admin.disableTable(TableName.valueOf(namespaceName, tableName));
            admin.deleteTable(TableName.valueOf(namespaceName, tableName));

            System.err.println("删表" + namespaceName + ":" + tableName + ",表删除成功!");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }finally {
            if (admin != null){
                try {
                    admin.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}*/
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * @author WEIYUNHUI
 * @date 2024/7/31 15:27
 *
 *  Connection
 *     Table
 *     Admin
 *
 */
public class HBaseUtil {


    /**
     * 获取AsyncConnection的方法
     */
    public static AsyncConnection getAsyncConnection(){
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        try {
            AsyncConnection asyncConnection = ConnectionFactory.createAsyncConnection(conf).get();
            return asyncConnection ;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("获取 Hbase AsyncConnection 失败");
        }
    }

    /**
     * 关闭AsyncConnection的方法
     */
    public static void closeAsyncConnection(AsyncConnection asyncConnection){
        if(asyncConnection != null && !asyncConnection.isClosed()){
            try {
                asyncConnection.close();
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException("关闭 Hbase AsyncConnection 失败");
            }
        }
    }

    public static <T> T getAsyncRow(AsyncConnection asyncConnection, String namespaceName, String tableName, String rowKey, Class<T> clz, Boolean isUnderToCamel){
        // 获取Table对象
        TableName tn = TableName.valueOf(namespaceName, tableName);
        AsyncTable<AdvancedScanResultConsumer> table = null;
        try {

            table = asyncConnection.getTable(tn);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get).get();
            // 获取所有的cell
            List<Cell> cells = result.listCells();
            // 返回的对象
            T t = clz.newInstance();

            // 迭代每个cell
            for (Cell cell : cells) {
                // 列名
                String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                // 列值
                String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                if (isUnderToCamel){
                    // 下划线转驼峰
                    columnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, columnName);
                }
                // 为对象属性赋值
                BeanUtils.setProperty(t, columnName, columnValue);

            }
            return t;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }



    /**
     * 获取Connection的方法
     */
    public static Connection getConnection(){
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        try {
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection ;
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("获取 Hbase Connection 失败");
        }
    }


    /**
     * 关闭Connection的方法
     */
    public static void closeConnection(Connection connection){
        if(connection != null && !connection.isClosed()){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException("关闭 Hbase Connection 失败");
            }
        }
    }

    /**
     * delete
     * @param connection
     * @param namespaceName
     * @param tableName
     * @param rk
     */
    public static void deleteRow(Connection connection , String namespaceName, String tableName , String rk ){
        Table table = null ;
        try {
            TableName tn = TableName.valueOf(namespaceName, tableName);
            table = connection.getTable(tn) ;
            Delete delete = new Delete(Bytes.toBytes(rk));
            table.delete(delete);
            System.err.println(" 删除 " + namespaceName+ ":" + tableName + " 表, rowkey为 " + rk+ " 的数据成功");
        }catch ( Exception e ){
            e.printStackTrace();
            throw new RuntimeException(" 删除 Hbase 失败 ");
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }


    /**
     * put
     * shell : put 'namespace:table' , 'rk' , 'cf:cl' , 'value'
     * @param connection
     * @param namespaceName
     * @param tableName
     * @param rk
     * @param cf
     * @param data  一条维度数据
     */
    public static void putRow(Connection connection , String namespaceName, String tableName , String rk , String cf , JSONObject data){
        //判空处理
        Table table = null ;
        try {
            //获取Table对象
            TableName tn = TableName.valueOf(namespaceName, tableName);
            table = connection.getTable(tn);

            Put put = new Put(Bytes.toBytes(rk));
            //指定列
            data.entrySet()
                    .stream()
                    .filter( entry -> entry.getValue() != null )  //将列值为空的列过滤掉
                    .forEach( entry -> put.addColumn(Bytes.toBytes(cf) , Bytes.toBytes(entry.getKey()) , Bytes.toBytes(entry.getValue().toString())));
            table.put( put );
            System.err.println(" 写入 " + namespaceName+ ":" + tableName + " 表, rowkey为 " + rk+ " 的数据成功");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(" 写入 Hbase 失败 ");
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 建表
     */
    public static void createTable(Connection connection , String namespaceName , String tableName, String cf ){
        // 非空判断
        Admin admin = null ;
        try {
            admin = connection.getAdmin();
            //1. 判断表是否已经存在
            TableName tn = TableName.valueOf(namespaceName, tableName);
            if(admin.tableExists( tn )){
                System.err.println(" 建表 "+ namespaceName+ ":" + tableName + " , 表已经存在");
                return ;
            }

            //2.建表
            //处理列族
            //ColumnFamilyDescriptor columnFamilyDescriptor =
            //       ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build();

            ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes(cf));

            //
            TableDescriptor tableDescriptor =
                    TableDescriptorBuilder
                            .newBuilder(tn)
                            .setColumnFamily(columnFamilyDescriptor)
                            .build();
            admin.createTable(tableDescriptor);

            System.err.println(" 建表 "+ namespaceName+ ":" + tableName + " , 表创建成功");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }finally {
            if(admin != null ){
                try {
                    admin.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * 删表
     */

    public static void dropTable( Connection connection , String namespaceName , String tableName ){
        // 非空判断
        Admin admin = null ;
        try {
            admin = connection.getAdmin();
            //1. 判断表是否已经存在
            TableName tn = TableName.valueOf(namespaceName, tableName);
            if(! admin.tableExists( tn )){
                System.err.println(" 删表 "+ namespaceName+ ":" + tableName + " , 表不存在");
                return ;
            }

            //2.删表
            admin.disableTable(tn);
            admin.deleteTable(tn);

            System.err.println(" 删表 " + namespaceName+ ":" + tableName + " , 表删除成功");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }finally {
            if(admin != null ){
                try {
                    admin.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     *  @param connection
     *  @param namespaceName   Hbase中维度库
     *  @param tableName  维度表
     *  @param rowKey  读取数据的rowkey
     *  @param clz  封装的返回值类型
     *  @param isUnderToCamel 是否进行下划线和驼峰的转换
     *  @return
     *  @param <T>
     */
    public static <T> T getRow(Connection connection, String namespaceName, String tableName, String rowKey, Class<T> clz, Boolean isUnderToCamel){
        // 获取Table对象
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = null;
        try {

             table = connection.getTable(tn);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            // 获取所有的cell
            List<Cell> cells = result.listCells();
            // 返回的对象
            T t = clz.newInstance();

            // 迭代每个cell
            for (Cell cell : cells) {
                // 列名
                String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                // 列值
                String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                if (isUnderToCamel){
                    // 下划线转驼峰
                    columnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, columnName);
                }
                // 为对象属性赋值
                BeanUtils.setProperty(t, columnName, columnValue);

            }
            return t;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }finally {
            try {
                if (table != null){
                    table.close();
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

    }
    public static void main(String[] args) {

        Connection connection = HBaseUtil.getConnection();
        JSONObject row = getRow(connection, Constant.HBASE_NAMESPACE, "dim_sku_info", "9", JSONObject.class, false);
        System.out.println(row);
        HBaseUtil.closeConnection(connection);
    }

}